package 天气数据;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;


import java.io.*;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

public class MapTest extends Mapper<LongWritable, Text, Data, NullWritable> {
    Data k = new Data();
    Map<String, String> sky_status = new HashMap<String, String>();
    String status;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        URI[] uris = context.getCacheFiles();
        File file = new File(uris[0]);
        BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(file)));
//        URI[] uris = context.getCacheFiles();
//        FileSystem fileSystem = FileSystem.get(uris[0], context.getConfiguration());
//        FSDataInputStream inputStream = fileSystem.open(new Path(uris[0]));
//        BufferedReader br = new BufferedReader(new InputStreamReader(inputStream));
        String line;
        while ((line = br.readLine()) != null) {
            sky_status.put(line.split(",")[0], line.split(",")[1]);
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String datas[] = value.toString().split("\\s+");
        if (datas.length != 12
                || Integer.parseInt(datas[7]) < 0
                || Integer.parseInt(datas[6]) < 0
                || (Integer.parseInt(datas[9]) < 0 || Integer.parseInt(datas[9]) > 10)
                || (Integer.parseInt(datas[5]) < 0 || Integer.parseInt(datas[5]) > 100)
                || (Integer.parseInt(datas[4]) < -40 || Integer.parseInt(datas[4]) > 50)) {
            return;
        }
        status = sky_status.get(datas[9]);
        k.set(datas[0], datas[1], datas[2], datas[3], datas[4], datas[5], Integer.parseInt(datas[6]), datas[7], datas[8], status, datas[10], datas[11]);
        context.write(k, NullWritable.get());
    }
}
